package com.mjf.file

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.api.common.serialization.{BulkWriter, SimpleStringEncoder}
import org.apache.flink.core.fs.{FSDataOutputStream, Path}
import org.apache.flink.core.io.SimpleVersionedSerializer
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer
import org.apache.flink.streaming.api.functions.sink.filesystem.{BucketAssigner, OutputFileConfig, StreamingFileSink}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object AddSinkForBulkFormat {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)

    val inputPath: String = AddSinkForBulkFormat.getClass.getResource("/input.txt").getPath
    val outputPath: String = "flink-share-connector/connector-file/output"

    // 从文本文件读取数据
    val source: DataStream[String] = env.readTextFile(inputPath)

    val result: DataStream[(String, Int)] = source
      .flatMap(_.split("\\W+"))
      .map(word => (word.toLowerCase, 1))
      .keyBy(_._1)
      .sum(1)

    result.print()

    // 写出文件
    result.addSink(
      StreamingFileSink.forBulkFormat[(String, Int)](new Path(outputPath), new BulkWriter.Factory[(String, Int)]() {
        override def create(out: FSDataOutputStream): BulkWriter[(String, Int)] = new BulkWriter[(String, Int)] {
          override def addElement(element: (String, Int)): Unit = element

          override def flush(): Unit = ???

          override def finish(): Unit = ???
        }
      })
//        .withBucketAssigner(new BasePathBucketAssigner())  // 指定桶（分区）分配策略为没有桶，默认为yyyy-MM-dd--HH
        .withOutputFileConfig(
          OutputFileConfig.builder()
            .withPartPrefix(AddSinkForBulkFormat.getClass.getSimpleName) // 文件前缀，默认是part
            .withPartSuffix("test") // part-0-0后缀，默认为空
            .build()
        )
        .build()
    )

    env.execute(AddSinkForBulkFormat.getClass.getName)

  }
}
